package com.kevin.eshopcache.constants;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Controller;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;

//@Controller
//@EnableAutoConfiguration
public class SampleController {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());
    /**
     * 注入kafkaTemplate
     */
    //@Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    /**
     * 发送消息的方法
     *
     * @param key  推送数据的key
     * @param data 推送数据的data
     */
    private void send(String key, String data) {



        ListenableFuture<SendResult<String, String>> test = kafkaTemplate.send("test", key, data);
        test.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                logger.info("[消息发送失败]，key={},data={}",key,data);
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                logger.info("[消息发送成功]，key={},data={}",key,data);
            }
        });
    }

    //@RequestMapping("/kafka")
    public String testKafka() {
        int iMax = 6;
        for (int i = 1; i < iMax; i++) {
            send("key" + i, "data" + i);
        }
        return "success";
    }

    public static void main(String[] args) {
        SpringApplication.run(SampleController.class, args);
    }

    /**
     * 使用日志打印消息
     */

    //@KafkaListener(topics = "test")
    public void receive(ConsumerRecord<?, ?> consumer) {
        logger.info("{} - {}:{}", consumer.topic(), consumer.key(), consumer.value());
    }

}


